
#include <ttg/math.h>
#include <ttg/services.h>
#include <ttg/pipeline.h>

using namespace ttg::pipeline;

//Splits incoming count of throws into sub-counts, sum of which is equal to total count.
class Scatter :public Node<int, int>
{
	virtual void process(int data)
	{
		int part_count = (int)ttg::services::CPUAnalyzer::getRef()->getCoreCount() * 4;
		for (int i = 0; i < part_count - 1; i++)
			sendNext(data / part_count);
		sendNext(data - (data / part_count) * (part_count - 1));
	}
};

//Throws points and stores hits/misses.
class Thrower :public Node<int, std::pair<int, int>>
{
	virtual void process(int data)
	{
		int hits = 0;
		for (int i = 0; i < data; i++)
		{
			double x = ttg::math::getFloatRandom(-1.0f, 1.0f);
			double y = ttg::math::getFloatRandom(-1.0f, 1.0f);
			if (x * x + y * y <= 1.0)
				hits++;
		}
		sendNext(std::make_pair(data, hits));
	}
};

//Gathers information about hits and total count of throws.
class Gatherer :public Node<std::pair<int, int>, double>
{
	private:
		ttg::threads::AtomicInt throw_count;			//it can be used in diffrent threads simultaneously
		ttg::threads::AtomicInt hits;					//due to its atomicity.
	public:
		//Clearing counters.
		virtual void onBegin()
		{
			throw_count.set(0);
			hits.set(0);
		}
		//Storing another results.
		virtual void process(std::pair<int, int> data)
		{
			throw_count.add(data.first);
			hits.add(data.second);
		}
		//Calculating Pi value.
		virtual void onEnd()
		{ sendNext((4.0 * hits.get()) / throw_count.get()); }
};

//Parallel version of Pi calculation (parallelized by ttg::pipeline)
double parallelPi(int throw_count)
{
	static								//avoids redundant pipeline creations
	Pipeline<int, double> pl(wrap(new Scatter()) + wrap(new Thrower()) + wrap(new Gatherer()));
	srand((int)ttg::services::Timer::getRef()->getMilliSeconds());
	return pl.run(throw_count)[0];
}
